package com.jiaying.sync;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;

import org.bson.Document;
import org.bson.types.ObjectId;

// 使用 Jackson 替换 jakarta.json 以提高性能
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonFactory.Feature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

/**
 * Kafka消息解析器
 * 用于解析Debezium MongoDB连接器生成的Kafka消息
 * 使用Jackson优化JSON解析性能，适用于JDK 25环境
 */
public class KafkaMessageParser {

    // 复用 ObjectMapper 和 JsonFactory 实例以减少对象创建开销
    private static final ObjectMapper objectMapper = new ObjectMapper();
    private static final JsonFactory jsonFactory = new JsonFactory();
    
    // 使用ConcurrentHashMap作为简单的对象池，适用于JDK 25
    private static final ConcurrentHashMap<Integer, Document> documentPool = new ConcurrentHashMap<>();
    
    // SQL模板缓存，避免重复生成相同表结构的SQL
    private static final ConcurrentHashMap<String, String> insertTemplates = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, String> updateTemplates = new ConcurrentHashMap<>();
    
    // 列名清理结果缓存，避免重复正则匹配
    private static final ConcurrentHashMap<String, String> sanitizedColumnNames = new ConcurrentHashMap<>();
    
    // 对象池的最大大小
    private static final int MAX_POOL_SIZE = 100;

    static {
        // 启用相关特性以提高性能
        jsonFactory.enable(Feature.CANONICALIZE_FIELD_NAMES);
        jsonFactory.enable(Feature.INTERN_FIELD_NAMES);
    }

    /**
     * Document资源包装类，支持自动归还到池中
     */
    private static class PooledDocument implements AutoCloseable {
        private final Document document;
        private boolean closed = false;
        
        public PooledDocument(Document document) {
            this.document = document;
        }
        
        public Document getDocument() {
            if (closed) {
                throw new IllegalStateException("Document has been returned to pool");
            }
            return document;
        }
        
        @Override
        public void close() {
            if (!closed) {
                returnDocument(document);
                closed = true;
            }
        }
    }
    
    /**
     * 从对象池获取Document实例
     * 使用JDK 25推荐的方式替代ThreadLocal
     * 
     * @return Document实例包装类
     */
    private static PooledDocument borrowPooledDocument() {
        // 使用当前线程ID的哈希值作为池键
        int key = Long.hashCode(Thread.currentThread().threadId());
        
        // 尝试从池中获取
        Document doc = documentPool.remove(key);
        
        if (doc == null) {
            // 如果池中没有，创建新的Document
            doc = new Document();
        } else {
            // 清空之前的内容
            doc.clear();
        }
        
        return new PooledDocument(doc);
    }
    
    /**
     * 归还Document实例到对象池
     * 
     * @param document 要归还的Document实例
     */
    private static void returnDocument(Document document) {
        if (document != null && documentPool.size() < MAX_POOL_SIZE) {
            int key = Long.hashCode(Thread.currentThread().threadId());
            documentPool.put(key, document);
        }
    }

    /**
     * 解析Kafka消息并提取关键信息（使用消息值和键）
     * 
     * @param messageKey   消息键（JSON格式）
     * @param messageValue 消息值（JSON格式）
     * @return 解析结果
     */
    public static ParsedMessage parseMessage(String messageKey, String messageValue) {
        try {
            // 使用Jackson解析JSON，避免重复解析
            JsonNode valueNode = objectMapper.readTree(messageValue);
            
            // 一次性提取所有需要的字段
            String operation = valueNode.path("op").asText("");
            JsonNode sourceNode = valueNode.path("source");
            String collectionName = sourceNode != null ? sourceNode.path("collection").asText("") : "";
            
            // 提取主键ID
            String id = extractId(valueNode, messageKey);

            // 根据操作符提取数据
            SqlStatement insertSqlStatement = null;
            SqlStatement updateSqlStatement = null;

            switch (operation) {
                case "c": // 创建
                case "r": // 读取（快照）
                    String afterJson = valueNode.path("after").asText(null);
                    if (afterJson != null) {
                        Document afterData = Document.parse(afterJson);
                        // 转换MongoDB特殊类型
                        convertDocumentTypes(afterData);
                        // 生成INSERT SQL语句
                        insertSqlStatement = generateInsertSQL(collectionName, afterData);
                    }
                    break;
                case "u": // 更新
                    JsonNode updateDescObj = valueNode.path("updateDescription");
                    if (updateDescObj != null) {
                        String updatedFields = updateDescObj.path("updatedFields").asText();
                        if (updatedFields != null) {
                            // 使用Jackson解析更新字段
                            JsonNode keyObject = objectMapper.readTree(updatedFields);
                            // 使用try-with-resources确保资源正确归还
                            try (PooledDocument pooledDoc = borrowPooledDocument()) {
                                Document updateDescription = pooledDoc.getDocument();
                                if (keyObject.isObject()) {
                                    ObjectNode objectNode = (ObjectNode) keyObject;
                                    objectNode.fieldNames().forEachRemaining(key -> {
                                        JsonNode value = objectNode.get(key);
                                        if (value != null && !value.isNull()) {
                                            Object convertedValue = convertMongoValue(value);
                                            updateDescription.put(key, convertedValue);
                                        }
                                    });
                                }
                                // 生成UPDATE SQL语句
                                updateSqlStatement = generateUpdateSQL(collectionName, id, updateDescription);
                            }
                        }
                    }
                    break;
                case "d": // 删除
                    // 删除操作通常没有after数据
                    break;
            }

            return new ParsedMessage(operation, collectionName, id, insertSqlStatement, updateSqlStatement);

        } catch (Exception e) {
            System.err.println("解析消息时出错: " + e.getMessage());
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 转换文档中的MongoDB特殊类型
     * 
     * @param document 要转换的文档
     */
    private static void convertDocumentTypes(Document document) {
        if (document == null) {
            return;
        }
        
        // 创建需要更新的字段列表，避免在迭代时修改集合
        List<String> fieldsToUpdate = new ArrayList<>(document.keySet());
        
        for (String key : fieldsToUpdate) {
            Object value = document.get(key);
            
            // 增加类型判断提前终止，避免不必要的递归
            if (value == null || value instanceof String || value instanceof Number || 
                value instanceof Boolean || value instanceof Date || value instanceof ObjectId ||
                value instanceof BigDecimal || value instanceof Pattern) {
                // 这些类型不需要转换，直接跳过
                continue;
            }
            
            if (value instanceof Document) {
                Document docValue = (Document) value;
                // 检查是否是MongoDB特殊类型
                Object convertedValue = convertSpecialTypes(docValue);
                if (convertedValue != null) {
                    document.put(key, convertedValue);
                } else {
                    // 嵌套文档，递归处理（但只处理一层深度）
                    convertDocumentTypesShallow(docValue);
                }
            } else if (value instanceof org.bson.types.Decimal128) {
                // 直接处理Decimal128类型
                org.bson.types.Decimal128 decimal128 = (org.bson.types.Decimal128) value;
                document.put(key, decimal128.bigDecimalValue());
            } else if (value instanceof org.bson.BsonRegularExpression) {
                // 处理BsonRegularExpression类型
                org.bson.BsonRegularExpression regex = (org.bson.BsonRegularExpression) value;
                Pattern pattern = Pattern.compile(regex.getPattern(), convertRegexOptions(regex.getOptions()));
                document.put(key, pattern);
            } else if (value instanceof List) {
                // 处理数组类型 - 使用快速序列化而不是toString()
                try {
                    String serializedList = objectMapper.writeValueAsString(value);
                    document.put(key, serializedList);
                } catch (Exception e) {
                    // 如果序列化失败，回退到toString()
                    document.put(key, value.toString());
                }
            }
        }
    }
    
    /**
     * 浅层转换文档中的MongoDB特殊类型（只处理一层深度）
     * 
     * @param document 要转换的文档
     */
    private static void convertDocumentTypesShallow(Document document) {
        if (document == null) {
            return;
        }
        
        // 创建需要更新的字段列表，避免在迭代时修改集合
        List<String> fieldsToUpdate = new ArrayList<>(document.keySet());
        
        for (String key : fieldsToUpdate) {
            Object value = document.get(key);
            
            // 增加类型判断提前终止，避免不必要的处理
            if (value == null || value instanceof String || value instanceof Number || 
                value instanceof Boolean || value instanceof Date || value instanceof ObjectId ||
                value instanceof BigDecimal || value instanceof Pattern) {
                // 这些类型不需要转换，直接跳过
                continue;
            }
            
            if (value instanceof Document) {
                Document docValue = (Document) value;
                // 检查是否是MongoDB特殊类型
                Object convertedValue = convertSpecialTypes(docValue);
                if (convertedValue != null) {
                    document.put(key, convertedValue);
                }
                // 不再递归处理嵌套文档
            } else if (value instanceof org.bson.types.Decimal128) {
                // 直接处理Decimal128类型
                org.bson.types.Decimal128 decimal128 = (org.bson.types.Decimal128) value;
                document.put(key, decimal128.bigDecimalValue());
            } else if (value instanceof org.bson.BsonRegularExpression) {
                // 处理BsonRegularExpression类型
                org.bson.BsonRegularExpression regex = (org.bson.BsonRegularExpression) value;
                Pattern pattern = Pattern.compile(regex.getPattern(), convertRegexOptions(regex.getOptions()));
                document.put(key, pattern);
            } else if (value instanceof List) {
                // 处理数组类型 - 使用快速序列化而不是toString()
                try {
                    String serializedList = objectMapper.writeValueAsString(value);
                    document.put(key, serializedList);
                } catch (Exception e) {
                    // 如果序列化失败，回退到toString()
                    document.put(key, value.toString());
                }
            }
        }
    }

    /**
     * 转换MongoDB值为适合SQL Server的类型
     * 
     * @param value 原始值
     * @return 转换后的值
     */
    private static Object convertMongoValue(Object value) {
        if (value == null) {
            return null;
        }
        
        // 处理特殊MongoDB类型
        Object converted = convertSpecialTypes(value);
        if (converted != null) {
            return converted;
        }
        
        // 处理基本类型
        if (value instanceof String || value instanceof Number || value instanceof Boolean) {
            return value;
        }
        
        // 处理日期类型
        if (value instanceof Date) {
            return new java.sql.Timestamp(((Date) value).getTime());
        }
        
        // 处理ObjectId类型
        if (value instanceof org.bson.types.ObjectId) {
            return value.toString();
        }
        
        // 处理列表类型
        if (value instanceof List) {
            // 转换为JSON字符串而不是直接使用toString()
            try {
                return objectMapper.writeValueAsString(value);
            } catch (Exception e) {
                // 如果JSON序列化失败，使用toString()
                return value.toString();
            }
        }
        
        // 处理Pattern类型（正则表达式）
        if (value instanceof Pattern) {
            return value.toString();
        }
        
        // 对于其他复杂类型，转换为JSON字符串
        if (value instanceof Document) {
            return ((Document) value).toJson();
        }
        
        // 默认情况下，转换为字符串
        return value.toString();
    }

    /**
     * 转换MongoDB值为适合SQL Server的类型（从JsonNode）
     * 
     * @param value JsonNode值
     * @return 转换后的值
     */
    private static Object convertMongoValue(JsonNode value) {
        if (value == null || value.isNull()) {
            return null;
        }
        
        // 处理特殊MongoDB类型
        Object converted = convertSpecialTypesFromJsonNode(value);
        if (converted != null) {
            return converted;
        }
        
        // 处理基本类型
        if (value.isTextual()) {
            return value.asText();
        } else if (value.isNumber()) {
            // 根据数值类型返回合适的Java类型
            if (value.isInt()) {
                return value.asInt();
            } else if (value.isLong()) {
                return value.asLong();
            } else if (value.isDouble()) {
                return value.asDouble();
            } else if (value.isFloat()) {
                return value.floatValue();
            } else if (value.isBigDecimal()) {
                return value.decimalValue();
            } else {
                return value.asDouble();
            }
        } else if (value.isBoolean()) {
            return value.asBoolean();
        }
        
        // 对于其他复杂类型，转换为JSON字符串
        return value.toString();
    }

    /**
     * 转换MongoDB特殊类型
     * 
     * @param value 包含特殊类型的Document
     * @return 转换后的值，如果不是特殊类型则返回null
     */
    private static Object convertSpecialTypes(Object value) {
        if (value instanceof Document) {
            Document docValue = (Document) value;
            if (docValue.containsKey("$oid")) {
                // ObjectId类型
                return docValue.getString("$oid");
            } else if (docValue.containsKey("$numberLong")) {
                // 64位长整数
                return Long.parseLong(docValue.getString("$numberLong"));
            } else if (docValue.containsKey("$numberInt")) {
                // 32位整数
                return Integer.parseInt(docValue.getString("$numberInt"));
            } else if (docValue.containsKey("$date")) {
                // Date类型 - 优化：直接返回时间戳字符串以减少对象创建开销
                Object dateValue = docValue.get("$date");
                if (dateValue instanceof Number) {
                    // 直接返回时间戳字符串而不是创建Date对象
                    return String.valueOf(((Number) dateValue).longValue());
                } else if (dateValue instanceof String) {
                    // 保持字符串格式
                    return dateValue;
                }
            } else if (docValue.containsKey("$timestamp")) {
                // Timestamp类型 - 保留为Document以便后续处理
                // 这里我们将其转换为字符串表示
                return docValue.toJson();
            } else if (docValue.containsKey("$binary")) {
                // Binary Data类型 - 优化：直接保留Base64字符串而不是转换为字节数组
                return docValue.getString("$binary");
            } else if (docValue.containsKey("$numberDecimal")) {
                // Decimal128类型 - 转换为BigDecimal
                String decimalValue = docValue.getString("$numberDecimal");
                return new BigDecimal(decimalValue);
            } else if (docValue.containsKey("$regex")) {
                // Regular Expression类型
                String regex = docValue.getString("$regex");
                String options = docValue.getString("$options");
                if (options == null) {
                    options = "";
                }
                return Pattern.compile(regex, convertRegexOptions(options));
            } else if (docValue.containsKey("$code")) {
                // JavaScript Code类型
                return docValue.getString("$code");
            }
        }
        
        // 不是特殊类型
        return null;
    }

    /**
     * 从JsonNode转换MongoDB特殊类型
     * 
     * @param value 包含特殊类型的JsonNode
     * @return 转换后的值，如果不是特殊类型则返回null
     */
    private static Object convertSpecialTypesFromJsonNode(JsonNode value) {
        if (value.isObject() && value instanceof ObjectNode) {
            ObjectNode objValue = (ObjectNode) value;
            if (objValue.has("$oid")) {
                // ObjectId类型
                return objValue.path("$oid").asText();
            } else if (objValue.has("$numberLong")) {
                // 64位长整数
                return objValue.path("$numberLong").asLong();
            } else if (objValue.has("$numberInt")) {
                // 32位整数
                return objValue.path("$numberInt").asInt();
            } else if (objValue.has("$date")) {
                // Date类型 - 优化：直接返回时间戳字符串以减少对象创建开销
                JsonNode dateValue = objValue.path("$date");
                if (dateValue.isNumber()) {
                    // 直接返回时间戳字符串而不是创建Date对象
                    return String.valueOf(dateValue.asLong());
                } else if (dateValue.isTextual()) {
                    // 保持字符串格式
                    return dateValue.asText();
                }
            } else if (objValue.has("$timestamp")) {
                // Timestamp类型 - 转换为字符串表示
                return objValue.toString();
            } else if (objValue.has("$binary")) {
                // Binary Data类型 - 优化：直接保留Base64字符串而不是转换为字节数组
                return objValue.path("$binary").asText();
            } else if (objValue.has("$numberDecimal")) {
                // Decimal128类型 - 转换为BigDecimal
                return new BigDecimal(objValue.path("$numberDecimal").asText());
            } else if (objValue.has("$regex")) {
                // Regular Expression类型
                String regex = objValue.path("$regex").asText();
                String options = objValue.path("$options").asText();
                if (options == null) {
                    options = "";
                }
                return Pattern.compile(regex, convertRegexOptions(options));
            } else if (objValue.has("$code")) {
                // JavaScript Code类型
                return objValue.path("$code").asText();
            }
        }
        
        // 不是特殊类型
        return null;
    }

    /**
     * 转换正则表达式选项
     * 
     * @param options MongoDB正则表达式选项字符串
     * @return Java Pattern标志
     */
    private static int convertRegexOptions(String options) {
        int flags = 0;
        if (options.contains("i")) {
            flags |= Pattern.CASE_INSENSITIVE;
        }
        if (options.contains("m")) {
            flags |= Pattern.MULTILINE;
        }
        if (options.contains("s")) {
            flags |= Pattern.DOTALL;
        }
        if (options.contains("x")) {
            flags |= Pattern.COMMENTS;
        }
        return flags;
    }

    /**
     * 从消息中提取主键ID
     * 
     * @param valueNode 消息JSON节点
     * @param messageKey 消息键
     * @return 主键ID
     */
    private static String extractId(JsonNode valueNode, String messageKey) {
        try {
            // 首先尝试从messageKey中提取ID（对于更新操作特别重要）
            if (messageKey != null && !messageKey.isEmpty()) {
                try {
                    JsonNode keyObject = objectMapper.readTree(messageKey);

                    JsonNode idValue = keyObject.path("id");
                    if (idValue != null && !idValue.isNull()) {
                        // ID可能是一个包含$oid的对象
                        if (idValue.isObject()) {
                            JsonNode oidString = idValue.path("$oid");
                            if (!oidString.isMissingNode()) {
                                return oidString.asText();
                            }
                            // 处理其他可能的ID格式
                            Object convertedId = convertMongoValue(idValue);
                            return convertedId.toString();
                        }
                        // 或者ID可能是一个JSON字符串（嵌套的JSON）
                        if (idValue.isTextual()) {
                            String idStr = idValue.asText();
                            // 尝试解析嵌套的JSON
                            try {
                                JsonNode nestedObject = objectMapper.readTree(idStr);
                                
                                // 检查嵌套对象是否包含$oid
                                if (nestedObject.has("$oid")) {
                                    return nestedObject.path("$oid").asText();
                                }
                                
                                // 如果没有$oid，递归处理嵌套对象
                                Object convertedId = convertMongoValue(nestedObject);
                                return convertedId.toString();
                            } catch (Exception e) {
                                // 如果不是有效的JSON，直接返回字符串
                                return idStr;
                            }
                        }
                        // 处理数字类型
                        if (idValue.isNumber()) {
                            return String.valueOf(idValue.asLong());
                        }
                        return idValue.toString();
                    }
                } catch (Exception e) {
                    // 如果messageKey不是有效的JSON，继续尝试其他方法
                    System.err.println("解析messageKey时出错: " + e.getMessage());
                }
            }

            // 尝试从after数据中提取ID（作为备选方案）
            String afterJson = valueNode.path("after").asText(null);
            if (afterJson != null) {
                Document afterDoc = Document.parse(afterJson);
                if (afterDoc.containsKey("_id")) {
                    Object idObj = afterDoc.get("_id");
                    // 使用类型转换方法处理ID
                    Object convertedId = convertIdValue(idObj);
                    return convertedId.toString();
                }
                // 处理新的ID格式 {"id": {"68f896a80ae338cca6441589"}, "name": "test3", "age": 23}
                else {
                    // 尝试解析整个after文档来查找id字段
                    Object idObj = afterDoc.get("id");
                    if (idObj != null) {
                        Object convertedId = convertIdValue(idObj);
                        return convertedId.toString();
                    }
                }
            }

            // 如果以上方法都失败，返回null
            return null;
        } catch (Exception e) {
            System.err.println("提取ID时出错: " + e.getMessage());
            return null;
        }
    }

    /**
     * 转换ID值为字符串
     * 
     * @param idObj ID对象
     * @return 转换后的ID字符串
     */
    private static Object convertIdValue(Object idObj) {
        if (idObj instanceof Document) {
            Document idDoc = (Document) idObj;
            if (idDoc.containsKey("$oid")) {
                return idDoc.getString("$oid");
            }
            if (idDoc.containsKey("$numberLong")) {
                return idDoc.getString("$numberLong");
            }
            if (idDoc.containsKey("$numberInt")) {
                return idDoc.getString("$numberInt");
            }
            // 对于其他文档格式，检查是否是嵌套的ID格式 {"id": {"$oid": "..."}}
            if (idDoc.containsKey("id") && idDoc.get("id") instanceof Document) {
                Document nestedIdDoc = (Document) idDoc.get("id");
                if (nestedIdDoc.containsKey("$oid")) {
                    return nestedIdDoc.getString("$oid");
                }
            }
            // 对于其他文档格式，返回其JSON字符串表示
            return idDoc.toJson();
        } else if (idObj instanceof Number) {
            // 处理数字类型的ID
            return String.valueOf(idObj);
        } else {
            // 处理字符串类型的ID
            return idObj.toString();
        }
    }

    /**
     * 解析结果类
     */
    public static class ParsedMessage {
        private final String operation;
        private final String collectionName;
        private final String id;
        private final SqlStatement insertSqlStatement;
        private final SqlStatement updateSqlStatement;

        public ParsedMessage(String operation, String collectionName, String id,
                SqlStatement insertSqlStatement, SqlStatement updateSqlStatement) {
            this.operation = operation;
            this.collectionName = collectionName;
            this.id = id;
            this.insertSqlStatement = insertSqlStatement;
            this.updateSqlStatement = updateSqlStatement;
        }

        // Getters
        public String getOperation() {
            return operation;
        }

        public String getCollectionName() {
            return collectionName;
        }

        public String getId() {
            return id;
        }

        public SqlStatement getInsertSqlStatement() {
            return insertSqlStatement;
        }

        public SqlStatement getUpdateSqlStatement() {
            return updateSqlStatement;
        }

        /**
         * 生成INSERT SQL语句及参数
         * 
         * @return SqlStatement对象，包含SQL语句和参数数组
         */
        public SqlStatement toInsertSQL() {
            if (!"c".equals(operation) && !"r".equals(operation)) {
                return null; // 只有创建和读取操作才生成INSERT语句
            }
            
            return insertSqlStatement;
        }

        /**
         * 生成UPDATE SQL语句及参数
         * 
         * @return SqlStatement对象，包含SQL语句和参数数组
         */
        public SqlStatement toUpdateSQL() {
            if (!"u".equals(operation)) {
                return null; // 只有更新操作才生成UPDATE语句
            }
            
            return updateSqlStatement;
        }

        @Override
        public String toString() {
            return "ParsedMessage{" +
                    "operation='" + operation + '\'' +
                    ", collectionName='" + collectionName + '\'' +
                    ", id='" + id + '\'' +
                    ", insertSqlStatement=" + insertSqlStatement +
                    ", updateSqlStatement=" + updateSqlStatement +
                    '}';
        }
    }

    /**
     * SQL语句和参数封装类
     */
    public static class SqlStatement {
        private final String sql;
        private final Object[] parameters;

        public SqlStatement(String sql, Object[] parameters) {
            this.sql = sql;
            this.parameters = parameters;
        }

        public String getSql() {
            return sql;
        }

        public Object[] getParameters() {
            return parameters;
        }

        @Override
        public String toString() {
            return "SqlStatement{sql='" + sql + "', parameters=" + java.util.Arrays.toString(parameters) + "}";
        }
    }

    /**
     * 生成UPSERT SQL语句及参数
     * 使用MERGE语句实现：如果记录存在则更新，否则插入
     * 
     * @param collectionName 集合名称
     * @param document       要插入的文档
     * @return SqlStatement对象，包含SQL语句和参数数组
     */
    public static SqlStatement generateInsertSQL(String collectionName, Document document) {
        if (document == null || document.isEmpty()) {
            return null;
        }

        // 预估StringBuilder容量以减少扩容开销
        int estimatedSize = 500 + document.size() * 50;
        StringBuilder columns = new StringBuilder(estimatedSize);
        java.util.List<Object> selectParameters = new java.util.ArrayList<>(); // 用于SELECT子句的参数

        String idValue = null;

        // 处理文档中的每个字段（除了_id）
        for (String key : document.keySet()) {
            Object value = document.get(key);
            
            // 跳过null值
            if (value == null) {
                continue;
            }
            
            // 提取ID值
            if ("_id".equals(key)) {
                // 处理ObjectId特殊格式以获取实际ID值
                if (value instanceof Document) {
                    Document docValue = (Document) value;
                    if (docValue.containsKey("$oid")) {
                        idValue = docValue.getString("$oid");
                    } else if (docValue.containsKey("$numberLong")) {
                        idValue = docValue.getString("$numberLong");
                    } else if (docValue.containsKey("$numberInt")) {
                        idValue = docValue.getString("$numberInt");
                    } else {
                        idValue = value.toString();
                    }
                } else if (value instanceof org.bson.types.ObjectId) {
                    idValue = value.toString();
                } else {
                    idValue = value.toString();
                }
                continue; // 不将_id添加到列中，因为我们将其转换为id
            }
            
            // 添加列名
            if (columns.length() > 0) {
                columns.append(", ");
            }
            
            String sanitizedColumnName = sanitizeColumnName(key);
            columns.append(sanitizedColumnName);
            
            // 添加参数到SELECT子句，确保类型适合SQL Server
            Object convertedValue = convertMongoValue(value);
            selectParameters.add(convertedValue);
        }

        // 添加ID列和值到SELECT参数列表的开头
        if (idValue != null) {
            selectParameters.add(0, idValue);
        }
        // 生成表名（按照项目规范）
        String tableName = collectionName;
        
        // 预估SQL构建器容量
        int sqlBuilderSize = 800 + document.size() * 100;
        StringBuilder sqlBuilder = new StringBuilder(sqlBuilderSize);
        sqlBuilder.append("MERGE ").append(tableName).append(" AS target ");
        sqlBuilder.append("USING (SELECT ");
        
        // 添加参数占位符
        if (idValue != null) {
            sqlBuilder.append("? AS id");
            if (columns.length() > 0) {
                sqlBuilder.append(", ");
            }
        }
        
        // 添加其他列的占位符
        int colIndex = 0;
        for (String key : document.keySet()) {
            if ("_id".equals(key) || document.get(key) == null) {
                continue;
            }
            
            if (colIndex > 0) {
                sqlBuilder.append(", ");
            }
            
            sqlBuilder.append("? AS ").append(sanitizeColumnName(key));
            colIndex++;
        }
        
        sqlBuilder.append(") AS source ");
        sqlBuilder.append("ON target.id = source.id ");
        sqlBuilder.append("WHEN MATCHED THEN ");
        sqlBuilder.append("UPDATE SET ");
        
        // 构建UPDATE SET子句
        colIndex = 0;
        for (String key : document.keySet()) {
            if ("_id".equals(key) || document.get(key) == null) {
                continue;
            }
            
            if (colIndex > 0) {
                sqlBuilder.append(", ");
            }
            String updateColumn = sanitizeColumnName(key);
            sqlBuilder.append(updateColumn).append(" = source.").append(updateColumn);
            colIndex++;
        }
        
        sqlBuilder.append(" ");
        sqlBuilder.append("WHEN NOT MATCHED THEN ");
        
        // 预估INSERT部分容量
        int insertBuilderSize = 100 + document.size() * 30;
        StringBuilder insertColumns = new StringBuilder(insertBuilderSize);
        StringBuilder insertValues = new StringBuilder(insertBuilderSize);
        
        if (idValue != null) {
            insertColumns.append("id");
            insertValues.append("source.id");
        }
        
        colIndex = 0;
        for (String key : document.keySet()) {
            if ("_id".equals(key) || document.get(key) == null) {
                continue;
            }
            
            if (insertColumns.length() > 0) {
                insertColumns.append(", ");
                insertValues.append(", ");
            }
            
            String cn = sanitizeColumnName(key);
            insertColumns.append(cn);
            insertValues.append("source.").append(cn);
            colIndex++;
        }
        
        sqlBuilder.append("INSERT (").append(insertColumns.toString()).append(") ");
        sqlBuilder.append("VALUES (").append(insertValues.toString()).append(");");
        
        // 合并参数：SELECT参数 + UPDATE参数
        java.util.List<Object> allParameters = new java.util.ArrayList<>();
        allParameters.addAll(selectParameters);
        
        return new SqlStatement(sqlBuilder.toString(), allParameters.toArray());
    }

    /**
     * 生成UPDATE SQL语句及参数
     * 
     * @param collectionName 集合名称
     * @param id             主键ID
     * @param updateDoc      更新描述文档
     * @return SqlStatement对象，包含SQL语句和参数数组
     */
    public static SqlStatement generateUpdateSQL(String collectionName, String id, Document updateDoc) {
        if (updateDoc == null || updateDoc.isEmpty() || id == null) {
            return null;
        }

        // 预估StringBuilder容量以减少扩容开销
        int estimatedSize = 200 + updateDoc.size() * 30;
        StringBuilder setClause = new StringBuilder(estimatedSize);
        java.util.List<Object> parameters = new java.util.ArrayList<>();

        // 处理更新的每个字段
        for (String key : updateDoc.keySet()) {
            Object value = updateDoc.get(key);
            
            // 跳过null值
            if (value == null) {
                continue;
            }
            
            // 添加SET子句
            if (setClause.length() > 0) {
                setClause.append(", ");
            }
            
            // 将_id字段改为id字段
            String columnName = key;
            if ("_id".equals(key)) {
                columnName = "id";
            }
            setClause.append(sanitizeColumnName(columnName)).append(" = ?");
            
            // 处理ObjectId特殊格式并确保类型适合SQL Server
            Object convertedValue = convertMongoValue(value);
            parameters.add(convertedValue);
        }

        // 如果没有要更新的字段，返回null
        if (setClause.length() == 0) {
            return null;
        }

        // 添加WHERE子句参数（ID）
        parameters.add(id);

        // 生成完整的UPDATE语句
        String sql = "UPDATE " + collectionName + " SET " + setClause.toString() + " WHERE id = ?";

        return new SqlStatement(sql, parameters.toArray());
    }
    
    /**
     * 获取UPDATE SQL模板
     * 
     * @param tableName 表名
     * @param columns 列名集合
     * @return SQL模板
     */
    private static String getUpdateTemplate(String tableName, Set<String> columns) {
        // 创建缓存键
        String key = tableName + ":" + String.join(",", columns);
        
        return updateTemplates.computeIfAbsent(key, k -> {
            // 生成UPDATE SQL模板
            int estimatedSize = 100 + columns.size() * 30;
            StringBuilder template = new StringBuilder(estimatedSize);
            template.append("UPDATE ").append(tableName).append(" SET ");
            
            // 添加SET子句
            boolean first = true;
            for (String column : columns) {
                if (!first) {
                    template.append(", ");
                }
                template.append(sanitizeColumnName(column)).append(" = ?");
                first = false;
            }
            
            template.append(" WHERE id = ?");
            return template.toString();
        });
    }

    /**
     * 从可能的复杂格式中提取实际的ID值
     * 
     * @param id 可能包含复杂格式的ID字符串
     * @return 提取后的实际ID值
     */
    private static String extractActualId(String id) {
        if (id == null || id.isEmpty()) {
            return id;
        }
        
        // 保存原始ID用于最终返回
        String originalId = id;
        
        try {
            // 递归处理嵌套的转义JSON字符串
            String processedId = id;
            while (processedId.startsWith("\"{") && processedId.endsWith("}\"")) {
                // 移除外层引号并替换转义引号
                processedId = processedId.substring(1, processedId.length() - 1).replace("\\\"", "\"");
            }
            
            // 尝试解析为JSON格式
            JsonNode jsonObject = objectMapper.readTree(processedId);
            
            // 检查是否有id字段
            JsonNode idValue = jsonObject.path("id");
            if (idValue != null && !idValue.isNull()) {
                // 如果id字段本身是一个对象，尝试提取$oid
                if (idValue.isObject()) {
                    JsonNode idObject = idValue;
                    JsonNode oidString = idObject.path("$oid");
                    if (!oidString.isMissingNode()) {
                        return oidString.asText();
                    }
                    // 如果没有$oid，但有其他键，返回第一个键
                    if (idObject.isObject() && idObject.size() > 0) {
                        return idObject.fieldNames().next();
                    }
                }
                // 如果id字段是字符串
                if (idValue.isTextual()) {
                    return idValue.asText();
                }
                // 如果id字段是数字
                if (idValue.isNumber()) {
                    return String.valueOf(idValue.asLong());
                }
                return idValue.toString();
            }
            
            // 检查是否有$oid字段（直接在根对象中）
            JsonNode oidString = jsonObject.path("$oid");
            if (!oidString.isMissingNode()) {
                return oidString.asText();
            }
            
            // 如果没有找到特殊字段，尝试递归解析processedId（如果它与原始ID不同）
            if (!processedId.equals(originalId)) {
                return extractActualId(processedId);
            }
            
            // 如果没有找到特殊字段，返回原始ID
            return originalId;
        } catch (Exception e) {
            // 如果不是有效的JSON格式，直接返回原始ID
            return originalId;
        }
    }

    /**
     * 清理列名，防止SQL注入
     * 使用缓存避免重复正则匹配
     * 
     * @param columnName 列名
     * @return 清理后的列名
     */
    private static String sanitizeColumnName(String columnName) {
        if (columnName == null || columnName.isEmpty()) {
            return columnName;
        }
        
        // 从缓存中获取清理后的列名
        return sanitizedColumnNames.computeIfAbsent(columnName, k -> {
            // 移除危险字符，只允许字母、数字、下划线
            return columnName.replaceAll("[^a-zA-Z0-9_]", "");
        });
    }

    /**
     * 获取INSERT SQL模板
     * 
     * @param tableName 表名
     * @param columns 列名集合
     * @return SQL模板
     */
    private static String getInsertTemplate(String tableName, Set<String> columns) {
        // 创建缓存键
        String key = tableName + ":" + String.join(",", columns);
        
        return insertTemplates.computeIfAbsent(key, k -> {
            // 生成INSERT SQL模板
            StringBuilder template = new StringBuilder();
            template.append("INSERT INTO ").append(tableName).append(" (");
            
            // 添加列名
            boolean first = true;
            for (String column : columns) {
                if (!first) {
                    template.append(", ");
                }
                template.append(sanitizeColumnName(column));
                first = false;
            }
            
            template.append(") VALUES (");
            
            // 添加参数占位符
            first = true;
            for (int i = 0; i < columns.size(); i++) {
                if (!first) {
                    template.append(", ");
                }
                template.append("?");
                first = false;
            }
            
            template.append(")");
            return template.toString();
        });
    }
    
    /**
     * 测试方法
     */
    public static void main(String[] args) {
        // 测试创建操作消息
        String createMessage = "{\"before\":null,\"after\":\"{\\\"_id\\\": {\\\"$numberLong\\\": \\\"1761208919156\\\"},\\\"idMg\\\": {\\\"$numberLong\\\": \\\"1761209919156\\\"},\\\"addtime\\\": {\\\"$date\\\": 1761208919142},\\\"content\\\": \\\"测试短信内容_14\\\",\\\"mGateway\\\": 4,\\\"mobileCount\\\": 26,\\\"msgCount\\\": 9,\\\"sendTime\\\": {\\\"$date\\\": 1761208919142},\\\"source\\\": 2,\\\"status\\\": -1,\\\"statusCode\\\": \\\"500572\\\",\\\"statusCodeDesc\\\": \\\"状态描述_14\\\",\\\"tGateway\\\": 1,\\\"eprId\\\": 68,\\\"ucuid\\\": \\\"user177\\\",\\\"uGateway\\\": 1,\\\"priority\\\": 0,\\\"md5Content\\\": \\\"md5_14\\\",\\\"ctxtype\\\": 1,\\\"platform\\\": 1,\\\"provinceOperator\\\": \\\"省份运营商_14\\\",\\\"tempID\\\": 647,\\\"contentType\\\": 4,\\\"taskId\\\": 9815,\\\"custTempId\\\": 709,\\\"timing\\\": 1,\\\"contentCount\\\": 11,\\\"label\\\": \\\"标签_14\\\",\\\"md5\\\": \\\"md5值_14\\\",\\\"existsLabel\\\": 0,\\\"hfType\\\": 1,\\\"onlyId\\\": {\\\"$numberLong\\\": \\\"1761210919156\\\"},\\\"msgLen\\\": 892,\\\"msgBytes\\\": 1081,\\\"toCheckPfoSameContent\\\": true,\\\"toCheckPfoSameTemplateId\\\": false,\\\"smsSign\\\": \\\"签名_14\\\"}\",\"updateDescription\":null,\"source\":{\"version\":\"3.3.1.Final\",\"connector\":\"mongodb\",\"name\":\"mongo-testdb\",\"ts_ms\":1761208917000,\"snapshot\":\"false\",\"db\":\"testdb\",\"sequence\":null,\"ts_us\":1761208917000000,\"ts_ns\":1761208917000000000,\"collection\":\"TBL_SMS_CONTENT_20251221\",\"ord\":2,\"lsid\":null,\"txnNumber\":null,\"wallTime\":1761208917705},\"op\":\"c\",\"ts_ms\":1761208917715,\"transaction\":null}";

        // 测试更新操作消息（使用您提供的实际格式，但简化为直接的消息格式）
        String updateMessage = "{\n" +
                "  \"before\": null,\n" +
                "  \"after\": null,\n" +
                "  \"updateDescription\": {\n" +
                "    \"removedFields\": null,\n" +
                "    \"updatedFields\": \"{\\\"name\\\": \\\"test4\\\"}\",\n" +
                "    \"truncatedArrays\": null\n" +
                "  },\n" +
                "  \"source\": {\n" +
                "    \"version\": \"3.3.1.Final\",\n" +
                "    \"connector\": \"mongodb\",\n" +
                "    \"name\": \"mongo-testdb\",\n" +
                "    \"ts_ms\": 1761113263000,\n" +
                "    \"snapshot\": \"false\",\n" +
                "    \"db\": \"testdb\",\n" +
                "    \"sequence\": null,\n" +
                "    \"ts_us\": 1761113263000000,\n" +
                "    \"ts_ns\": 1761113263000000000,\n" +
                "    \"collection\": \"users2\",\n" +
                "    \"ord\": 1,\n" +
                "    \"lsid\": null,\n" +
                "    \"txnNumber\": null,\n" +
                "    \"wallTime\": 1761113263634\n" +
                "  },\n" +
                "  \"op\": \"u\",\n" +
                "  \"ts_ms\": 1761113263770,\n" +
                "  \"transaction\": null\n" +
                "}";

        // 测试消息键，使用复杂ID格式
        String updateKey = "{\"id\":\"{\\\"$oid\\\":\\\"68f874a60ae338cca6441587\\\"}\"}";

        System.out.println("解析创建操作消息:");
        KafkaMessageParser.ParsedMessage createResult = KafkaMessageParser.parseMessage(updateKey, createMessage);
        if (createResult != null) {
            System.out.println("操作符: " + createResult.getOperation());
            System.out.println("表名: " + createResult.getCollectionName());
            System.out.println("主键ID: " + createResult.getId());
            
            // 生成INSERT SQL
            KafkaMessageParser.SqlStatement insertSQL = createResult.toInsertSQL();
            if (insertSQL != null) {
                System.out.println("生成的INSERT SQL: " + insertSQL.getSql());
                System.out.println("SQL参数: " + java.util.Arrays.toString(insertSQL.getParameters()));
            }
        }

        System.out.println("\n解析更新操作消息:");
        KafkaMessageParser.ParsedMessage updateResult = KafkaMessageParser.parseMessage(updateKey, updateMessage);
        if (updateResult != null) {
            System.out.println("操作符: " + updateResult.getOperation());
            System.out.println("表名: " + updateResult.getCollectionName());
            System.out.println("主键ID: " + updateResult.getId());
            
            // 生成UPDATE SQL
            KafkaMessageParser.SqlStatement updateSQL = updateResult.toUpdateSQL();
            if (updateSQL != null) {
                System.out.println("生成的UPDATE SQL: " + updateSQL.getSql());
                System.out.println("SQL参数: " + java.util.Arrays.toString(updateSQL.getParameters()));
            }
        }
        
        // 额外测试复杂ID提取
        System.out.println("\n测试复杂ID提取:");
        String complexId = "{\"id\":\"{\\\"$oid\\\":\\\"68f874a60ae338cca6441587\\\"}\"}";
        System.out.println("原始复杂ID: " + complexId);
        // 注意：这里我们直接调用私有方法用于测试，实际使用中不会这样调用
    }
}